package com.wfh.mianshiji.mq;


import cn.hutool.core.date.DateUnit;
import cn.hutool.core.date.DateUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.wfh.mianshiji.ai.agent.ResultAnalysisAgent;
import com.wfh.mianshiji.ai.model.StructedBean;
import com.wfh.mianshiji.common.ErrorCode;
import com.wfh.mianshiji.config.RabbitMqConfig;
import com.wfh.mianshiji.constant.RedisPrefixConstant;
import com.wfh.mianshiji.exception.BusinessException;
import com.wfh.mianshiji.model.dto.ChatMessageDto;
import com.wfh.mianshiji.model.dto.MqMessageDto;
import com.wfh.mianshiji.model.entity.Conversion;
import com.wfh.mianshiji.model.entity.Interview;
import com.wfh.mianshiji.model.entity.Result;
import com.wfh.mianshiji.model.enums.InterviewStatusEnum;
import com.wfh.mianshiji.service.ConversionService;
import com.wfh.mianshiji.service.InterviewService;
import com.rabbitmq.client.Channel;
import com.wfh.mianshiji.service.ResultService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @Title: Consumer
 * @Author wangfenghuan
 * @Package com.wfh.mianshiji.mq
 * @Date 2025/11/3 09:18
 * @description:
 */
@Slf4j
@Component
public class Consumer {

    @Resource
    private InterviewService interviewService;

    @Resource
    private ConversionService conversionService;

    @Resource
    private ObjectMapper objectMapper;

    @Resource
    private ResultService resultService;

    @Resource(name = "chatRedisTemplate")
    private RedisTemplate<String, String> redisTemplate;

    @Resource
    private ResultAnalysisAgent resultAnalysisAgent;

    /**
     * 消费者
     * @param mqMessageDto
     * @param channel
     * @param deliveryTag
     * @throws IOException
     */
    @RabbitListener(queues = {RabbitMqConfig.QUEUE_NAME}, ackMode = "MANUAL")
    public void receive(MqMessageDto mqMessageDto, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
        log.info("收到消息：{}", mqMessageDto.getInterviewId());
        Long userId = mqMessageDto.getUserId();
        Long interviewId = mqMessageDto.getInterviewId();
        String redisKey = RedisPrefixConstant.CHAT_HISTORY_PREFIX + interviewId;
        // 查询当前用户的当前面试场次的所有对话记录以及简历信息
        Interview interview = interviewService.getOne(new QueryWrapper<>(Interview.class)
                .eq("userId", userId)
                .eq("id", interviewId));
        // 1. 从 Redis 获取对话记录（忽略第一条：从 index=1 开始）
        List<String> jsonMessages = redisTemplate.opsForList().range(redisKey, 1, -1);
        if (jsonMessages == null || jsonMessages.isEmpty()) {
            log.info("Redis 中无有效对话记录，interviewId: {}", interviewId);
            return;
        }
        // 2. 解析 JSON 并过滤有效消息
        List<ChatMessageDto> validMessages = new ArrayList<>();
        for (String json : jsonMessages) {
            try {
                ChatMessageDto dto = objectMapper.readValue(json, ChatMessageDto.class);
                if ("user".equals(dto.getRole()) || "ai".equals(dto.getRole())) {
                    validMessages.add(dto);
                }
            } catch (Exception e) {
                log.warn("跳过无效消息: {}", json, e);
            }
        }
        // 3. 按 role 分组（类似你对 MySQL 的操作）
        Map<String, List<ChatMessageDto>> grouped = validMessages.stream()
                .collect(Collectors.groupingBy(ChatMessageDto::getRole));
        List<ChatMessageDto> user = grouped.getOrDefault("user", Collections.emptyList());
        List<ChatMessageDto> ai = grouped.getOrDefault("ai", Collections.emptyList());
        // 面试状态判断
        if (!interview.getInterviewStatus().equals(InterviewStatusEnum.FINISHED.getValue())) {
            // 拒绝消息
            try {
                channel.basicNack(deliveryTag, false, false);
            } catch (IOException e) {
                throw new BusinessException(ErrorCode.SYSTEM_ERROR, "消息拒绝失败");
            }
        }
        // 获取当场面试的面试信息
        String jobTitle = interview.getJobTitle();
        String jobDescription = interview.getJobDescription();
        Date startTime = interview.getStartTime();
        Date stopTime = interview.getStopTime();
        long duration = DateUtil.between(startTime, stopTime, DateUnit.MINUTE);
        // 更新面试状态为进行中
        Result result = new Result();
        result.setUserId(userId);
        result.setInterviewId(interviewId);
        result.setInterviewType(0);
        result.setJobTitle(jobTitle);
        result.setStatus(1);
        result.setInterviewDuration(duration);
        resultService.save(result);
        QueryWrapper<Result> queryWrapper = new QueryWrapper<>(Result.class).eq("userId", userId).eq("interviewId", interviewId);
        String prompt = String.format("这是用户的面试岗位名称：%s" +
                "这是用户的面试岗位职责: %s +" +
                "这是当前面试的面试官问题: %s" +
                "这是当前面试的对应的求职者对应的回答：%s" +
                "请根据当前的内容按照要求给我分析面试情况", jobTitle, jobDescription, ai.toString(), user.toString());
        // 调用面试分析agent
        StructedBean structedBean = null;
        try {
            structedBean = resultAnalysisAgent.doChat(prompt, String.valueOf(interviewId));
        } catch (Exception e) {
            // 分析失败，拒绝消息，并更新生成状态
            handleErr(3, channel, result, queryWrapper, deliveryTag);
            throw new BusinessException(ErrorCode.SYSTEM_ERROR, "分析失败");
        }
        if (structedBean == null){
            // 分析失败，拒绝消息，并更新生成状态
            handleErr(3, channel, result, queryWrapper, deliveryTag);
            throw new BusinessException(ErrorCode.SYSTEM_ERROR, "分析失败");
        }
        // 获取面试分析结果
        String echartsCode = structedBean.getEchartsCode();
        String skills = structedBean.getSkills();
        String suggestions = structedBean.getSuggestions();
        String comment = structedBean.getComment();
        Integer interviewGrade = structedBean.getInterviewGrade();
        // 保存面试结果,并更新状态
        result.setInterviewGrade(interviewGrade);
        result.setInterviewDuration(duration);
        result.setEcharts(echartsCode);
        result.setSuggestion(suggestions);
        result.setComment(comment);
        result.setSkill(skills);
        result.setStatus(2);
        // 保存到数据库,并更新状态
        boolean update = resultService.update(result, queryWrapper);
        if (!update){
            // 保存失败，
            log.error("面试：{},结果保存失败", interviewId);
            handleErr(3, channel, result, queryWrapper, deliveryTag);
        }

        // 保存面试分析之后，将Redis中的面试记录同步到MySQL
        saveToDb(interviewId, userId);
        // 确认消息
        channel.basicAck(deliveryTag, false);
        log.info("面试：{}分析完成",interviewId);
    }

    /**
     * 保存对话记录到MySQL，并删除Redis中的对话记录
     *
     * @param interviewId 面试场次ID（雪花ID）
     */
    private void saveToDb(Long interviewId, Long userId) {
        if (interviewId == null) {
            log.warn("interviewId 为空，跳过保存");
            return;
        }

        String conversationId = interviewId.toString();
        String redisKey = RedisPrefixConstant.CHAT_HISTORY_PREFIX + conversationId;

        try {
            // 2. 从 Redis 获取所有消息（使用与 InterviewRedisChatMemory 相同的反序列化逻辑）
            List<String> jsonMessages = redisTemplate.opsForList().range(redisKey, 0, -1);
            if (jsonMessages == null || jsonMessages.isEmpty()) {
                log.info("Redis 中无对话记录，interviewId: {}", interviewId);
                // 仍尝试删除 key 避免残留
                redisTemplate.delete(redisKey);
                return;
            }

            // 3. 构建 Conversion 实体列表
            List<Conversion> conversions = new ArrayList<>();
            ObjectMapper objectMapper = new ObjectMapper();

            for (String json : jsonMessages) {
                try {
                    ChatMessageDto dto = objectMapper.readValue(json, ChatMessageDto.class);
                    Conversion conv = new Conversion();
                    conv.setInterviewId(interviewId);
                    conv.setUserId(userId);
                    conv.setMessage(dto.getContent());
                    conv.setMessageType("user".equals(dto.getRole()) ? "user" : "ai");
                    conversions.add(conv);
                } catch (Exception e) {
                    log.warn("跳过无效消息: {}", json, e);
                }
            }

            if (conversions.isEmpty()) {
                log.warn("无有效消息可保存，interviewId: {}", interviewId);
                redisTemplate.delete(redisKey);
                return;
            }

            // 4. 【幂等关键】先删除旧记录，再插入新记录
            conversionService.remove(new QueryWrapper<Conversion>().eq("interviewId", interviewId).eq("userId", userId));
            conversionService.saveBatch(conversions);

            log.info("对话已保存到 MySQL，共 {} 条消息，interviewId: {}", conversions.size(), interviewId);

            // 5. 成功后删除 Redis 中的对话记录（释放内存）
            redisTemplate.delete(redisKey);
            log.debug("Redis 对话记录已清理: {}", redisKey);

        } catch (Exception e) {
            log.error("保存对话到 MySQL 失败，interviewId: {}", interviewId, e);
            // 注意：这里不删除 Redis，避免数据丢失！可后续补偿
            // 可选：发送告警 or 记录失败任务表
        }
    }

    /**
     * 处理错误
     * @param status
     * @param channel
     * @param result
     * @param queryWrapper
     * @param deliveryTag
     * @throws IOException
     */
    private void handleErr(Integer status, Channel channel, Result result, QueryWrapper<Result> queryWrapper, long deliveryTag) throws IOException {
        // 分析失败，拒绝消息，并更新生成状态
        result.setStatus(status);
        resultService.update(result, queryWrapper);
        channel.basicNack(deliveryTag, false, false);
    }

}
